//package com.netvine.transform.security;
//
//import com.fasterxml.jackson.databind.ObjectMapper;
//import com.netvine.constant.Constant;
//import com.netvine.constant.LevelEnum;
//import com.netvine.entity.clean.count.OverViewA;
//import com.netvine.entity.clean.count.OverViewB;
//import com.netvine.entity.clean.denoise.NewDealWithDenoiseDto;
//import com.netvine.sink.overview.OverViewASecurityAlarmMysqlSink;
//import com.netvine.utils.DateUtil;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.flink.api.common.state.MapState;
//import org.apache.flink.api.common.state.MapStateDescriptor;
//import org.apache.flink.api.common.typeinfo.Types;
//import org.apache.flink.api.java.functions.KeySelector;
//import org.apache.flink.api.java.tuple.Tuple1;
//import org.apache.flink.api.java.tuple.Tuple5;
//import org.apache.flink.configuration.Configuration;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
//import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
//import org.apache.flink.streaming.api.windowing.evictors.Evictor;
//import org.apache.flink.streaming.api.windowing.time.Time;
//import org.apache.flink.streaming.api.windowing.triggers.ContinuousProcessingTimeTrigger;
//import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
//import org.apache.flink.streaming.runtime.operators.windowing.TimestampedValue;
//import org.apache.flink.util.Collector;
//
//import java.io.IOException;
//import java.io.Serializable;
//import java.util.*;
//import java.util.stream.Collectors;
//import java.util.stream.StreamSupport;
//
///**
// * @Author: wangzhou
// * @Date: 2024/10/25 10:21:23
// * @Description:
// **/
//@Slf4j
//public class InputApplyEventOverView implements Serializable {
//
//    private final ObjectMapper objectMapper = new ObjectMapper();
//
//    private static MapStateDescriptor<String, Integer> statisticsOverViewACountDesc =
//            new MapStateDescriptor<String, Integer>("statistics_overviewa_count", Types.STRING, Types.INT);
//
//    private static MapStateDescriptor<String, Integer> statisticsOverViewBCountDesc =
//            new MapStateDescriptor<String, Integer>("statistics_overviewb_count", Types.STRING, Types.INT);
//
//    private static MapStateDescriptor<String, String> statisticsOverViewIPsDesc =
//            new MapStateDescriptor<String, String>("statistics_overviewIps_collect", Types.STRING, Types.STRING);
//
//    private static MapStateDescriptor<String, Integer> statisticsAlertScopeDesc =
//            new MapStateDescriptor<String, Integer>("statistics_alert_scope_collect", Types.STRING, Types.INT);
//
//
//    public void applyA(SingleOutputStreamOperator<String> denoiseData) {
//        denoiseData
//                .map(x -> objectMapper.readValue(x, NewDealWithDenoiseDto.class))
//                .keyBy(new KeySelector<NewDealWithDenoiseDto, Tuple1<String>>() {
//                    @Override
//                    public Tuple1<String> getKey(NewDealWithDenoiseDto value) throws Exception {
//                        return Tuple1.of(value.alertScope);
//                    }
//                })
//                .window(TumblingProcessingTimeWindows.of(Time.hours(24), Time.hours(16)))
//                .allowedLateness(Time.hours(16))
//                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(30)))
//                .evictor(new Evictor<NewDealWithDenoiseDto, TimeWindow>() {
//                    @Override
//                    public void evictBefore(Iterable<TimestampedValue<NewDealWithDenoiseDto>> elements, int size, TimeWindow window, EvictorContext evictorContext) {
//
//                    }
//
//                    @Override
//                    public void evictAfter(Iterable<TimestampedValue<NewDealWithDenoiseDto>> elements, int size, TimeWindow window, EvictorContext evictorContext) {
//                        Iterator<TimestampedValue<NewDealWithDenoiseDto>> iterator = elements.iterator();
//                        while (iterator.hasNext()) {
//                            iterator.next();
//                            iterator.remove();
//                        }
//                    }
//                })
//                .process(new ProcessWindowFunction<NewDealWithDenoiseDto, OverViewA, Tuple1<String>, TimeWindow>() {
//
//                    MapState<String, Integer> statisticsOverViewACountState = null;
//
//                    MapState<String, String> statisticsOverViewAIpsState = null;
//
//                    MapState<String, Integer> statisticsAlertScopeCountState = null;
//
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        super.open(parameters);
//                        statisticsOverViewACountState = getRuntimeContext().getMapState(statisticsOverViewACountDesc);
//                        statisticsOverViewAIpsState = getRuntimeContext().getMapState(statisticsOverViewIPsDesc);
//                        statisticsAlertScopeCountState = getRuntimeContext().getMapState(statisticsAlertScopeDesc);
//
//
//
//                    }
//
//                    @Override
//                    public void process(Tuple1<String> tuple1, ProcessWindowFunction<NewDealWithDenoiseDto,
//                            OverViewA, Tuple1<String>, TimeWindow>.Context context,
//                                        Iterable<NewDealWithDenoiseDto> elements, Collector<OverViewA> out) throws Exception {
//                        log.info(" >>>> InputApplyEventOverViewA 开始计算：[" + DateUtil.LongTimeStampToDateTime(System.currentTimeMillis()) + "] ");
//                        String mapKey = DateUtil.toDateTime(context.window().getStart() / 1000)
//                                + "-" + tuple1.f0;
//                        String lowMapkey= mapKey + "-" + LevelEnum.LOW_LEVEL.getCode();
//                        String mediumMapkey= mapKey + "-"  + LevelEnum.MEDIUM_LEVEL.getCode();
//                        String highMapkey= mapKey + "-" +  LevelEnum.HIGH_LEVEL.getCode();
//                        String criMapkey= mapKey + "-" + LevelEnum.CRIT_LEVEL.getCode();
//                        String collectIpsMapKey = mapKey + Constant.COLLECT_IPSKEY_POSTFIX;
//                        int elementCount = 0;
//                        int lowelCount = 0;
//                        int mediumCount = 0;
//                        int highCount = 0;
//                        int criCount = 0;
//                        List<NewDealWithDenoiseDto> newDealWithDenoiseDtoList = StreamSupport.stream(elements.spliterator(), false).collect(Collectors.toList());
//                        String collectJoinOrinIpStr = newDealWithDenoiseDtoList.stream().map(NewDealWithDenoiseDto::getDstIp).distinct().collect(Collectors.joining(","));
//                        if (statisticsOverViewAIpsState.contains(collectIpsMapKey)) {
//                            String collectHistoryCollect= statisticsOverViewAIpsState.get(collectIpsMapKey);
//                            collectJoinOrinIpStr = collectJoinOrinIpStr + "," + collectHistoryCollect;
//                            collectJoinOrinIpStr = Arrays.stream(collectJoinOrinIpStr.split(",")).distinct().collect(Collectors.joining(","));
//                            statisticsOverViewAIpsState.put(collectIpsMapKey, collectJoinOrinIpStr);
//                        } else {
//                            statisticsOverViewAIpsState.put(collectIpsMapKey, collectJoinOrinIpStr);
//                        }
//
//
//                        try {
//
//                            String mapAlertKey = DateUtil.toDateTime(context.window().getStart() / 1000);
//
//                            if (statisticsAlertScopeCountState.contains(mapAlertKey + "-" + 2)) {
//                                Integer i = statisticsAlertScopeCountState.get(mapAlertKey + "-" + 2);
//                                log.info("这个区域的降噪后的数据是:{}",i);
//                            }
//
//
//                            elementCount = newDealWithDenoiseDtoList.size();
//                            List<Integer> integerList = newDealWithDenoiseDtoList.stream().map(x -> x.getEventLevel()).collect(Collectors.toList());
//                            for (Integer i : integerList) {
//                                switch (i) {
//                                    case 1:
//                                        criCount++;
//                                        break;
//                                    case 2:
//                                        highCount++;
//                                        break;
//                                    case 3:
//                                        mediumCount++;
//                                        break;
//                                    case 4:
//                                        lowelCount++;
//                                        break;
//                                }
//                            }
//                            elementCount = getElementCount(statisticsOverViewACountState,mapKey, elementCount);
//                            lowelCount = getElementCount(statisticsOverViewACountState,lowMapkey, lowelCount);
//                            mediumCount = getElementCount(statisticsOverViewACountState,mediumMapkey, mediumCount);
//                            highCount = getElementCount(statisticsOverViewACountState,highMapkey, highCount);
//                            criCount = getElementCount(statisticsOverViewACountState,criMapkey, criCount);
//
//
//
//                            // 删除 statisticsOverViewACountState 2天之前的数据
//                            Iterator<Map.Entry<String, Integer>> iteratorStates = statisticsOverViewACountState.iterator();
//                            while (iteratorStates.hasNext()) {
//                                Map.Entry<String, Integer> next = iteratorStates.next();
//                                long deleteTimeStamp = context.window().getStart() - com.netvine.constant.Configuration.TWO_TIME_STAMP;
//                                String mapKeyDelete = DateUtil.toDateTime(deleteTimeStamp / 1000);
//                                if (next.getKey().startsWith(mapKeyDelete)) {
//                                    iteratorStates.remove();
//                                }
//                            }
//                            // 删除 statisticsOverViewAIpsState 2天之前的数据
//                            Iterator<Map.Entry<String, String>> iteratorIpsState = statisticsOverViewAIpsState.iterator();
//                            while (iteratorIpsState.hasNext()) {
//                                Map.Entry<String, String> next = iteratorIpsState.next();
//                                long deleteTimeStamp = context.window().getStart() - com.netvine.constant.Configuration.TWO_TIME_STAMP;
//                                String mapKeyDelete = DateUtil.toDateTime(deleteTimeStamp / 1000);
//                                if (next.getKey().startsWith(mapKeyDelete)) {
//                                    iteratorIpsState.remove();
//                                }
//                            }
//
//                            newDealWithDenoiseDtoList.stream().map(x -> x.getEventLevel().equals(LevelEnum.CRIT_LEVEL.getCode())).count();
//
//                            OverViewA result = new OverViewA();
//                            result.setDt(DateUtil.LongTimeStampToDate(context.window().getStart()));
//                            // TODO: 配合降噪那边进行做
//                            result.setSolveTotal(-1);// 降噪后的告警数
//                            result.setTotal(elementCount);
//                            result.setLowAlertCount(lowelCount);
//                            result.setMediumAlertCount(mediumCount);
//                            result.setHighAlertCount(highCount);
//                            result.setCritAlertCount(criCount);
//                            result.setAlertScope(tuple1.f0);
//
//                            result.setDstIps(collectJoinOrinIpStr);
//                            log.info("告警概览A写入数据的结果:{}", result);
//                            out.collect(result);
//                        } catch (IOException e) {
//                            log.error("InputApplyEventOverViewA error! {}", e.getMessage());
//                        }
//                    }
//                })
//                .disableChaining().name("InputApplyEventOverViewA sink window")
//                .addSink(new OverViewASecurityAlarmMysqlSink().eventResultSinkA())
//                .name("InputApplyEventOverViewA sink mysql");
//
//    }
//
//    private static int getElementCount(MapState<String,Integer> state,String mapKey, int count) throws Exception {
//        if (state.contains(mapKey)) {
//            Integer size = state.get(mapKey);
//            count += size;
//            state.put(mapKey, count);
//        } else {
//            state.put(mapKey, count);
//        }
//        return count;
//    }
//
//    public void applyB(SingleOutputStreamOperator<String> denoiseData) {
//        denoiseData
//                .map(x -> objectMapper.readValue(x, NewDealWithDenoiseDto.class))
//                .keyBy(new KeySelector<NewDealWithDenoiseDto, Tuple5<String,String,String,String,String>>() {
//                    @Override
//                    public Tuple5<String,String,String,String,String> getKey(NewDealWithDenoiseDto value) throws Exception {
//                        return Tuple5.of(value.dstIp,value.eventType,value.equipment,value.alertScope,value.protocol);
//                    }
//                })
//                .window(TumblingProcessingTimeWindows.of(Time.hours(24), Time.hours(16)))
//                .allowedLateness(Time.hours(16))
//                .trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(30)))
//                .evictor(new Evictor<NewDealWithDenoiseDto, TimeWindow>() {
//                    @Override
//                    public void evictBefore(Iterable<TimestampedValue<NewDealWithDenoiseDto>> elements, int size, TimeWindow window, EvictorContext evictorContext) {
//
//                    }
//
//                    @Override
//                    public void evictAfter(Iterable<TimestampedValue<NewDealWithDenoiseDto>> elements, int size, TimeWindow window, EvictorContext evictorContext) {
//                        Iterator<TimestampedValue<NewDealWithDenoiseDto>> iterator = elements.iterator();
//                        while (iterator.hasNext()) {
//                            iterator.next();
//                            iterator.remove();
//                        }
//                    }
//                })
//                .process(new ProcessWindowFunction<NewDealWithDenoiseDto, OverViewB, Tuple5<String,String,String,String,String>, TimeWindow>() {
//
//
//
//                    MapState<String, Integer> statisticsOverViewBCountState = null;
//                    MapState<String, Integer> statisticsAlertScopeCountState = null;
//
//                    @Override
//                    public void open(Configuration parameters) throws Exception {
//                        super.open(parameters);
//                        statisticsOverViewBCountState = getRuntimeContext().getMapState(statisticsOverViewBCountDesc);
//                        statisticsAlertScopeCountState = getRuntimeContext().getMapState(statisticsAlertScopeDesc);
//                    }
//
//                    @Override
//                    public void process(Tuple5<String, String, String, String, String> tuple5, ProcessWindowFunction<NewDealWithDenoiseDto,
//                            OverViewB, Tuple5<String, String, String, String, String>, TimeWindow>.Context context,
//                                        Iterable<NewDealWithDenoiseDto> elements, Collector<OverViewB> out) throws Exception {
//                        log.info(" >>>> InputApplyEventOverViewB 开始计算：[" + DateUtil.LongTimeStampToDateTime(System.currentTimeMillis()) + "] ");
//                        String mapKey = DateUtil.toDateTime(context.window().getStart() / 1000)
//                                + "-" + tuple5.f0 + tuple5.f1 + tuple5.f2 + tuple5.f3 + tuple5.f4;
//                        String mapAlertKey = DateUtil.toDateTime(context.window().getStart() / 1000);
//                        int elementCount = 0;
//                        List<NewDealWithDenoiseDto> newDealWithDenoiseDtoList = StreamSupport.stream(elements.spliterator(), false).collect(Collectors.toList());
//
//                        try {
//                            statisticsAlertScopeCountState = getRuntimeContext().getMapState(statisticsAlertScopeDesc);
//
//
//                            newDealWithDenoiseDtoList.stream().map(x -> mapAlertKey + "-" + x.getAlertScope()).distinct()
//                                    .forEach(x->{
//                                        try {
//                                            if (statisticsAlertScopeCountState.contains(x)) {
//                                                Integer integer = statisticsAlertScopeCountState.get(x);
//                                                integer++;
//                                                statisticsAlertScopeCountState.put(x, integer);
//                                            } else {
//                                                statisticsAlertScopeCountState.put(x, 1);
//                                            }
//                                        } catch (Exception e) {
//                                            log.error("statisticsAlertScopeCountState 处理数据出现错误,错误信息是:{}",e.getMessage());
//                                        }
//                                    });
//                            elementCount = newDealWithDenoiseDtoList.size();
//                            if (statisticsOverViewBCountState.contains(mapKey)) {
//                                Integer size = statisticsOverViewBCountState.get(mapKey);
//                                elementCount += size;
//                                statisticsOverViewBCountState.put(mapKey, elementCount);
//                            } else {
//                                statisticsOverViewBCountState.put(mapKey, elementCount);
//                            }
//                            Iterator<Map.Entry<String, Integer>> iteratorStates = statisticsOverViewBCountState.iterator();
//
//                            while (iteratorStates.hasNext()) {
//                                Map.Entry<String, Integer> next = iteratorStates.next();
//                                long deleteTimeStamp = context.window().getStart() - com.netvine.constant.Configuration.TWO_TIME_STAMP;
//                                String mapKeyDelete = DateUtil.toDateTime(deleteTimeStamp / 1000);
//                                if (next.getKey().startsWith(mapKeyDelete)) {
//                                    iteratorStates.remove();
//                                }
//                            }
//                            OverViewB result = new OverViewB();
//                            result.setDt(DateUtil.LongTimeStampToDate(context.window().getStart()));
//                            result.setTotal(elementCount);
//                            result.setDstIp(tuple5.f0);
//                            result.setEventType(tuple5.f1);
//                            result.setEquipment(tuple5.f2);
//                            result.setAlertScope(tuple5.f3);
//                            result.setProtocol(tuple5.f4);
//                            log.info("告警概览B写入数据的结果:{}", result);
//                            out.collect(result);
//                        } catch (IOException e) {
//                            log.error("InputApplyEventOverViewB error! {}", e.getMessage());
//                        }
//                    }
//                })
//                .disableChaining().name("InputApplyEventOverViewB sink mysql  window")
//                .addSink(new OverViewASecurityAlarmMysqlSink().eventResultSinkB())
//                .name("InputApplyEventOverViewB sink mysql");
//
//    }
//
//
//}
